package arthur.dy.lee.tair.impl;

import arthur.dy.lee.tair.LockCallback;
import arthur.dy.lee.tair.Locker;
import arthur.dy.lee.tair.TairLockCallback;
import arthur.dy.lee.tair.UnLockException;
import com.taobao.tair.ResultCode;
import com.taobao.tair.TairManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

/**
 * Created by Administrator on 2017/8/21.
 */
public class TairLocker<T, K> implements Locker<T, K> {

    private static final Logger logger = LoggerFactory.getLogger(TairLocker.class);

    private static final int DEFAULT_EXPIRE_TIME = 30 * 60;

    private static final int DEFAULT_TAIR_LOCK_VERSION = 100;

    private TairManager tairManager;
    private int         namespace;

    public T doInLock(LockCallback<T, K> callback, K k) throws Exception {
        Assert.notNull("callback must not be null");
        Assert.isInstanceOf(TairLockCallback.class, callback, "callback must instance of TairLockCallback");

        TairLockCallback<T, K> tairLockCallback = (TairLockCallback<T, K>) callback;
        String key = tairLockCallback.buildKey(k);
        try {
            lock(key);
            return callback.invoke(k);
        } finally {
            try {
                unlock(key);
            } catch (Exception e) {
                logger.error("unlock in doSafeInLock error:" + e.getMessage(), e);
            }
        }
    }

    public T doSafeInLock(LockCallback<T, K> callback, K k, int expireTime) throws Exception {
        Assert.notNull("callback must not be null");
        Assert.isInstanceOf(TairLockCallback.class, callback, "callback must instance of TairLockCallback");

        TairLockCallback<T, K> tairLockCallback = (TairLockCallback<T, K>) callback;
        String key = tairLockCallback.buildKey(k);
        lock(key, expireTime);
        try {
            T t = callback.invoke(k);
            return t;
        } finally {
            try {
                unlock(key);
            } catch (Exception e) {
                logger.error("unlock in doSafeInLock error:" + e.getMessage(), e);
            }
        }
    }

    private void lock(String key, int expireTime) throws Exception {
        ResultCode resultCode = this.tairManager.put(this.namespace, key, "", DEFAULT_TAIR_LOCK_VERSION, expireTime);
        if (!resultCode.isSuccess()) {
            logger.error("lock failed, code:{},msg:{}", new Object[] { resultCode.getCode(), resultCode.getMessage() });
            throw new UnLockException("locked failed");
        }
    }

    private void lock(String key) throws Exception {
        ResultCode resultCode = this.tairManager.put(this.namespace, key, "", DEFAULT_TAIR_LOCK_VERSION,
                DEFAULT_EXPIRE_TIME);
        if (!resultCode.isSuccess()) {
            logger.error("lock failed, code:{},msg:{}", new Object[] { resultCode.getCode(), resultCode.getMessage() });
            throw new RuntimeException("locked failed");
        }
    }

    /**
     * 有超时时间，最后执行一下，其实无所谓了这个操作
     *
     * @param key
     * @throws Exception
     */
    private void unlock(String key) throws Exception {
        ResultCode resultCode = this.tairManager.invalid(this.namespace, key);
        if (!resultCode.isSuccess()) {
            logger.error("unlock failed, code:{},msg:{}",
                    new Object[] { resultCode.getCode(), resultCode.getMessage() });
        }
    }

    public void setTairManager(TairManager tairManager) {
        this.tairManager = tairManager;
    }

    public void setNamespace(int namespace) {
        this.namespace = namespace;
    }
}
